package com.zl.rockemqsource.example.simplemsg;

import com.zl.rockemqsource.constant.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

/**
 * @author zhouliang
 * @date 2020/03/22
 **/
@Slf4j
public class SyncProducer {

    public static void main(String[] args)
            throws MQClientException {
        //设置唯一的生产者组，每个实例的生产者组唯一
        DefaultMQProducer producer = new DefaultMQProducer(RocketMQConstant.PRODUCER_GROUP);
        //设置nameserver的地址
        producer.setNamesrvAddr(RocketMQConstant.NAMESRV_ADDR);
        //启动实例
        producer.start();

//        producer.createTopic("broker-a", "TopicTest", 4);

        try {
            for (int i = 0; i < 100000; i++) {
                Message msg = new Message(
                        "TopicTest",
                        "TagA",
                        ("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
                );
                SendResult sendResult = producer.send(msg);

                log.info("同步发送消息返回:{}", sendResult);
            }
        } catch (UnsupportedEncodingException | InterruptedException | MQBrokerException | RemotingException | MQClientException e) {
            log.error("", e);
        } finally {
            //一旦不再使用生产者的时候，请关闭它
            producer.shutdown();
        }
    }
}
